package com.zyx.flinkdemo.sql.tableapi;

import com.zyx.flinkdemo.pojo.WaterSensor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author zyx
 * @since 2021/5/23 07:19
 * desc: 将流转化为表并进行聚合运算
 */
public class StreamToTableAggDemo {
    public static void main(String[] args) throws Exception {
        // 创建流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 读取端口数据创建流并转化为WaterSensor Pojo
        SingleOutputStreamOperator<WaterSensor> waterSensorDs = env
                .socketTextStream("linux201", 8888)
                .map(data -> {
                    String[] split = data.split(",");
                    return new WaterSensor(split[0],
                            Long.parseLong(split[1]),
                            Integer.parseInt(split[2]));
                });

        // 创建表的执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 将流转化为动态表
        Table sensorTable = tableEnv.fromDataStream(waterSensorDs);

        // 对表中数据进行筛选并聚合
        Table groupTable = sensorTable
                .where($("vc").isGreaterOrEqual(16))
                .groupBy($("id"))
                .aggregate($("vc").sum().as("vc_sum"))
                .select($("id"), $("vc_sum"));

        /*// 过时写法
        Table groupTable = sensorTable
                .groupBy("id")
                .select("id, vc.sum");*/

        // 将动态表转化为回撤流并进行输出
        DataStream<Tuple2<Boolean, Row>> resDataStream = tableEnv.toRetractStream(groupTable, Row.class);
        resDataStream.print();

        // 执行任务
        env.execute();

    }
}
